package gateway.zk;

import gateway.Constants;
import gateway.ServicesHolder;
import gateway.WebContext;
import gateway.plugins.divide.LoadBalance;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.web.reactive.context.ReactiveWebServerInitializedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Objects;

@Slf4j
@Service
public class ServiceRegisterAndDiscoverImpl implements ServiceRegisterAndDiscover,
        //Webflux的web容器启动事件，不是tomcat
        ApplicationListener<ReactiveWebServerInitializedEvent> {
    @Autowired
    private ZkClient zkClient;
    private String registryPath = Constants.ZK_REGISTRY;
    @Override
    public void onApplicationEvent(ReactiveWebServerInitializedEvent event) {
        //根路径的注册和服务路径的注册是容器初始化的时候就完成的
        //因为有哪些服务是已经确定的，所以这两个方法是私有的
        registryRootPath();
        registryServicePath();
        //此刻还没有任何一个真实的address注册进来
        //调用zookeeper就不展开讲了，可以和图中对照
    }
    private void registryRootPath() {
        if (!zkClient.exists(registryPath)) {
            zkClient.createPersistent(registryPath);
            log.info(">>> create registry node:" + registryPath);
        }
    }
    private void registryServicePath(){
        ServicesHolder.getServiceNames()
            .stream()
            .map(ServicesHolder::addServicePath)
            .filter(servicePath -> !zkClient.exists(servicePath))
            .peek(servicePath -> {
                //创建service节点（持久）
                zkClient.createPersistent(servicePath);
                log.info(">>>create service node:" + servicePath);
            });
    }
    @Override//根据服务名注册服务具体的地址，这个方法是对外暴露的
    public void registryAddress(String serviceName, String address){
        String servicePath = ServicesHolder.addAddress(serviceName, address);
        String addressNode = zkClient.createPersistentSequential(servicePath, address);
        log.info(">>> create address node:" + addressNode);
    }
    @Override
    public String discover(WebContext webContext, LoadBalance loadBalance) {
        String serviceName = webContext.getServiceName();
        String servicePath = ServicesHolder.getServicePath(serviceName);
        //获取服务节点
        if (!zkClient.exists(servicePath)) {
            throw new RuntimeException(String.format(">>>can't find any service node on path {}",servicePath));
        }
        //从本地缓存获取某个服务地址
        //不同的负载策略获取地址的算法不一样，后续
        String address = loadBalance.load(webContext);
        if (Objects.isNull(address)){
            List<String> addressList = zkClient.getChildren(servicePath);
            ServicesHolder.addAddress(servicePath, addressList);
            //监听servicePath下的子文件是否发生变化，如果发生变化就要刷新注册中心缓存
            zkClient.subscribeChildChanges(servicePath,(parentPath,currentChilds)->{
                log.info(">>>servicePath is changed:" + parentPath);
                ServicesHolder.addAddress(servicePath, currentChilds);
            });
            if (CollectionUtils.isEmpty(addressList)) {
                throw new RuntimeException(String.format(">>>can't find any address node on path {}", servicePath));
            }
            address = loadBalance.load(webContext);
            log.info(">>>get address node:" + address);
        }
        return address;
    }
}
